﻿using System;
using System.Text;
using Newtonsoft.Json;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace Infrastructure.Utility.Core.RabbitMq.Providers
{
    public class Subscriber : ISubscriber
    {
        private readonly IProvider _provider;
        public Subscriber(IProvider provider)
        {
            _provider = provider;
        }

        public void Subscribe<T>(IConnection connection, Action<T> callback) where T : class
        {
            Subscribe(connection, _provider.QueueName, _provider.RouteKey, callback);
        }
        public void Subscribe<T>(IConnection connection, string queueName, string routeKey, Action<T> callback) where T : class
        {
            IModel channel = null;
            try
            {
                channel = _provider.CreateChannel(connection);
                channel.ExchangeDeclare(_provider.ExchangeName,
                    ExchangeType.Topic,
                    _provider.Durable,
                    false,
                    null);
                //声明队列
                channel.QueueDeclare(
                    queue: queueName,
                    durable: _provider.Durable,
                    exclusive: false,
                    autoDelete: false,
                    arguments: null);

                //将队列与交换机进行绑定
                channel.QueueBind(
                    queue: queueName,
                    exchange: _provider.ExchangeName,
                    routingKey: routeKey);

                //声明为手动确认，每次只消费1条消息。
                channel.BasicQos(0, 1, false);

                //定义消费者
                var consumer = new EventingBasicConsumer(channel);
                //接收事件
                consumer.Received += (eventSender, args) =>
                {
                    //接收到的消息
                    var message = Encoding.UTF8.GetString(args.Body.ToArray());
                    callback(JsonConvert.DeserializeObject<T>(message));
                    //返回消息确认
                    channel.BasicAck(args.DeliveryTag, true);
                };
                //开启监听
                channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer);
            }
            catch (Exception ex)
            {
                if (channel != null)
                {
                    if (channel.IsOpen) channel.Close();
                    channel.Abort();
                    channel.Dispose();
                }
                _provider.ResetConnectionToFree(connection);
            }
        }
    }
}
